package org.example.事务消息.监听;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * @Auther:Liu
 * @Testname:事务消息的监听
 * @Date:2025/8/23 9:50
 */
public class 事务消息的监听 implements TransactionListener {

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        System.out.println("开始确认消息：" + message);
        String keys = message.getKeys();
        switch (keys) {
            case "Num:1", "Num:3", "Num:5" -> {
                return LocalTransactionState.COMMIT_MESSAGE;
            }
            case "Num:2", "Num:4", "Num:6" -> {
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            default -> {
                return LocalTransactionState.UNKNOW;
            }
        }
    }

    private int i = 0;

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        System.out.println("开始其后消息的确认");
        String keys = messageExt.getKeys();
        i++;
        switch (keys) {
            case "Num:7", "Num:8" -> {
                return LocalTransactionState.COMMIT_MESSAGE;
            }
            case "Num:9" -> {
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
            case "Num:10" -> {
                if (i < 4) {
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        }
        return null;
    }
}
